package com.gitee.melin.bee.core.jdbc.dialect.impl;

import com.gitee.melin.bee.core.jdbc.dialect.AbstractJdbcDialect;
import com.gitee.melin.bee.core.jdbc.enums.DataSourceType;
import com.gitee.melin.bee.core.jdbc.relational.BeeTableType;
import com.gitee.melin.bee.core.jdbc.relational.ConnectionInfo;
import com.google.common.collect.Lists;
import java.sql.*;
import java.util.List;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SnowflakeJdbcDialect extends AbstractJdbcDialect {

    private static final Logger LOG = LoggerFactory.getLogger(SnowflakeJdbcDialect.class);

    private static final String QUERY_DATABASES_SQL = "SHOW DATABASES";

    public SnowflakeJdbcDialect(ConnectionInfo connectionInfo) {
        super(connectionInfo);
    }

    public SnowflakeJdbcDialect(Connection connection) {
        super(connection);
    }

    @Override
    public DataSourceType getDataSourceType() {
        return DataSourceType.SNOWFLAKE;
    }

    @Override
    protected Driver getDriver() {
        try {
            Class<?> clazz = Class.forName(DataSourceType.SNOWFLAKE.getDriverClass());
            return (Driver) clazz.getDeclaredConstructor().newInstance();
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    @Override
    protected boolean filterSchema(String schemaName) {
        return super.filterSchema(schemaName);
    }

    @Override
    protected String getDDLColumnName(BeeTableType type) {
        return "CT_SCRIPT";
    }

    @Override
    public List<String> getDatabases() {
        try (Connection connection = connectionFactory.openConnection();
                PreparedStatement preparedStatement = connection.prepareStatement(QUERY_DATABASES_SQL);
                ResultSet resultSet = preparedStatement.executeQuery()) {

            List<String> catalogs = Lists.newArrayList();
            while (resultSet.next()) {
                String catalogName = resultSet.getString("name");
                catalogs.add(catalogName);
            }
            return catalogs;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    @Override
    protected String getSelectTablesSql(String tableName) {
        String sql = "SELECT TABLE_SCHEMA,\n"
                + "       TABLE_NAME,\n"
                + "       TABLE_TYPE,\n"
                + "       ROW_COUNT    TABLE_ROWS,\n"
                + "       BYTES        DATA_SIZE,\n"
                + "       CREATED      CREATE_TIME,\n"
                + "       LAST_ALTERED UPDATE_TIME,\n"
                + "       COMMENT      TABLE_COMMENT\n"
                + "FROM INFORMATION_SCHEMA.TABLES\n"
                + "WHERE TABLE_SCHEMA = :schemaName";

        if (StringUtils.isNotBlank(tableName)) {
            sql += " AND c.relnam = :tableName";
        }
        return sql;
    }

    @Override
    protected String createTableSql(BeeTableType type, String schemaName, String tableName) {
        if (BeeTableType.TABLE == type) {
            return String.format("select get_ddl('table', '%s.%s') as CT_SCRIPT", schemaName, tableName);
        } else {
            return String.format("select get_ddl('view', '%s.%s') as CT_SCRIPT", schemaName, tableName);
        }
    }
}
